package com.twoandone.tms.realtime.app.dwd;

import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdWayBillBound {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_waybill_bound"));
        //TODO 运单表
        Table order_info = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_no'] order_no,\n" +
                "data['status'] status,\n" +
                "data['collect_type'] collect_type,\n" +
                "data['user_id'] user_id,\n" +
                "data['receiver_complex_id'] receiver_complex_id,\n" +
                "data['receiver_province_id'] receiver_province_id,\n" +
                "data['receiver_city_id'] receiver_city_id,\n" +
                "data['receiver_district_id'] receiver_district_id,\n" +
                "data['receiver_address'] receiver_address,\n" +
                "data['receiver_name'] receiver_name,\n" +
                "data['receiver_phone'] receiver_phone,\n" +
                "data['receive_location'] receive_location,\n" +
                "data['sender_complex_id'] sender_complex_id,\n" +
                "data['sender_province_id'] sender_province_id,\n" +
                "data['sender_city_id'] sender_city_id,\n" +
                "data['sender_district_id'] sender_district_id,\n" +
                "data['sender_address'] sender_address,\n" +
                "data['sender_name'] sender_name,\n" +
                "data['sender_phone'] sender_phone,\n" +
                "data['send_location'] send_location,\n" +
                "data['payment_type'] payment_type,\n" +
                "data['cargo_num'] cargo_num,\n" +
                "data['amount'] amount,\n" +
                "data['estimate_arrive_time'] estimate_arrive_time,\n" +
                "data['distance'] distance,\n" +
                "ts,\n" +
                "proc_time\n" +
                "from `topic_db` \n" +
                "where `table` = 'order_info'\n" +
                "and `type`= 'insert'");
        tableEnv.createTemporaryView("order_info", order_info);
        //tableEnv.executeSql("select * from order_info").print();
        //TODO  运单机构中转表
        Table order_org_bound = tableEnv.sqlQuery("select \n" +
                "data['id'] id ,\n" +
                "data['order_id'] order_id ,\n" +
                "data['org_id'] org_id ,\n" +
                "data['status'] status ,\n" +
                "data['inbound_time'] inbound_time ,\n" +
                "data['inbound_emp_id'] inbound_emp_id ,\n" +
                "data['sort_time'] sort_time ,\n" +
                "data['sorter_emp_id'] sorter_emp_id ,\n" +
                "data['outbound_time'] outbound_time ,\n" +
                "data['outbound_emp_id'] outbound_emp_id ,\n" +
                "data['create_time'] create_time ,\n" +
                "ts,\n" +
                "proc_time\n" +
                "from `topic_db`\n" +
                "where `table` = 'order_org_bound' and \n" +
                "`type` = 'insert'");
        tableEnv.createTemporaryView("order_org_bound", order_org_bound);
        //tableEnv.executeSql("select * from order_org_bound").print();
        //TODO 获取字典表
        tableEnv.executeSql(MySqlUtil.getBaseDicLookUpDDL());
        // tableEnv.executeSql("select * from base_dic").print();
        //TODO 将运单表与运单机构中转表以及字典表关联
        Table joinTable = tableEnv.sqlQuery("select \n" +
                "oi.id,\n" +
                "oi.order_no,\n" +
                "oi.status,\n" +
                "oi.collect_type,\n" +
                "oi.user_id,\n" +
                "oi.receiver_complex_id,\n" +
                "oi.receiver_province_id,\n" +
                "oi.receiver_city_id,\n" +
                "oi.receiver_district_id,\n" +
                "oi.receiver_address,\n" +
                "oi.receiver_name,\n" +
                "oi.receiver_phone,\n" +
                "oi.receive_location,\n" +
                "oi.sender_complex_id,\n" +
                "oi.sender_province_id,\n" +
                "oi.sender_city_id,\n" +
                "oi.sender_district_id,\n" +
                "oi.sender_address,\n" +
                "oi.sender_name,\n" +
                "oi.sender_phone,\n" +
                "oi.send_location,\n" +
                "oi.payment_type,\n" +
                "oi.cargo_num,\n" +
                "oi.amount,\n" +
                "oi.estimate_arrive_time,\n" +
                "oi.distance,\n" +
                "oob.org_id,\n" +
                "oob.status,\n" +
                "oob.inbound_time,\n" +
                "oob.inbound_emp_id,\n" +
                "oob.sort_time,\n" +
                "oob.sorter_emp_id,\n" +
                "oob.outbound_time,\n" +
                "oob.outbound_emp_id,\n" +
                "oob.create_time,\n" +
                "oi.ts,\n" +
                "oidic.name as oi_status_name, \n" +
                "oiidic.name as collect_type_name, \n" +
                "oobdic.name as oob_status_name \n" +
                "from order_info oi " +
                "join " +
                "order_org_bound oob\n" +
                "on oi.id = oob.order_id\n" +
                "join " +
                "`base_dic` for system_time as of oi.proc_time as oidic\n" +
                "on cast(oi.status as bigint) = oidic.id \n" +
                "join " +
                "`base_dic` for system_time as of oi.proc_time as oiidic\n" +
                "on cast(oi.collect_type as bigint) = oiidic.id \n" +
                "join " +
                "`base_dic` for system_time as of oob.proc_time as oobdic\n" +
                "on cast(oob.status as bigint) = oobdic.id \n");
        tableEnv.createTemporaryView("join_table", joinTable);
        // tableEnv.executeSql("select * from join_table").print();
        //创建动态表与kafka主题进行映射
        tableEnv.executeSql("create table dwd_waybill_bound (\n" +
                "id string,\n" +
                "order_no string,\n" +
                "oistatus string,\n" +
                "collect_type string,\n" +
                "user_id string,\n" +
                "receiver_complex_id string,\n" +
                "receiver_province_id string,\n" +
                "receiver_city_id string,\n" +
                "receiver_district_id string,\n" +
                "receiver_address string,\n" +
                "receiver_name string,\n" +
                "receiver_phone string,\n" +
                "receive_location string,\n" +
                "sender_complex_id string,\n" +
                "sender_province_id string,\n" +
                "sender_city_id string,\n" +
                "sender_district_id string,\n" +
                "sender_address string,\n" +
                "sender_name string,\n" +
                "sender_phone string,\n" +
                "send_location string,\n" +
                "payment_type string,\n" +
                "cargo_num string,\n" +
                "amount string,\n" +
                "estimate_arrive_time string,\n" +
                "distance string,\n" +
                "org_id string,\n" +
                "status string,\n" +
                "inbound_time string,\n" +
                "inbound_emp_id string,\n" +
                "sort_time string,\n" +
                "sorter_emp_id string,\n" +
                "outbound_time string,\n" +
                "outbound_emp_id string,\n" +
                "create_time string,\n" +
                "ts string,\n" +
                "oi_status_name string,\n" +
                "collect_type_name string,\n" +
                "oob_status_name string,\n" +
                "primary key (id) not enforced\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_waybill_bound"));
        tableEnv.executeSql("insert into dwd_waybill_bound select * from join_table");
    }
}
